package com.cn.esermis.server;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.cn.esermis.config.RedisQueueService;
import com.cn.esermis.constant.Constant;
import com.cn.esermis.dpld.business.MonitorBusiness;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import java.util.concurrent.*;

/**
 * 消费者类
 */
@Component
public class Consumer {

    @Autowired
    private MonitorBusiness monitorBusiness;
    @Autowired
    private RedisQueueService redisQueueService;


    @Async
    public void receive(){
        System.out.println("开始消费消息");
        // 开启线程池，如果有数据就启动线程消费
        final int[] tolot = {0};

        int corePoolSize = 20; // 核心线程数
        int maximumPoolSize = 50; // 最大线程数
        long keepAliveTime = 60L; // 空闲线程存活时间（秒）
        TimeUnit unit = TimeUnit.SECONDS; // 时间单位
        BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>(200); // 工作队列
        ThreadFactory threadFactory = Executors.defaultThreadFactory(); // 线程工厂
        RejectedExecutionHandler handler = new ThreadPoolExecutor.AbortPolicy(); // 拒绝策略
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                corePoolSize,
                maximumPoolSize,
            keepAliveTime,
            unit,
            workQueue,
            threadFactory,
            handler
        );
        while (true) {
            synchronized(this){
                //取出消息
                boolean message = redisQueueService.isQueueEmpty(Constant.SERVER);
                if (message) {
                    System.out.println("消息队列为空");
                    try {
                        Thread.sleep(1000 * 5);
                    } catch (InterruptedException e) {
                    }
                } else {

//                    tolot[0]++;
//                    int finalTolot = tolot[0];
//                    Object mge = redisQueueService.receiveMessage(Constant.SERVER);
//                    try {
//                        System.out.println("消费消息------------：第（" + finalTolot + "）条");
//                        JSONObject jsonObject = (JSONObject) JSON.toJSON(mge);
//                        monitorBusiness.inboundOperations(jsonObject);
//                    } catch (Exception e) {
//                        throw new RuntimeException(e);
////                        redisQueueService.sendMessage(Constant.RECEIVEREPEAT, mge);
//                    }
                    // 检查工作队列是否已满
                    if (workQueue.size() >= workQueue.remainingCapacity()) {
                        System.out.println("工作队列已满，无法提交新任务");
                        // 处理队列已满的情况，例如记录日志或采取其他措施
                    } else {
                        executor.submit(() -> {
                            Object mge = redisQueueService.receiveMessage(Constant.SERVER);
                            if (mge == null) {
                                return;
                            }
                            tolot[0]++;
                            int finalTolot = tolot[0];
                            try {
                                System.out.println("消费消息------------：第（" + finalTolot + "）条");
                                JSONObject jsonObject = (JSONObject) JSON.toJSON(mge);
//                                System.out.println(jsonObject);
                                monitorBusiness.inboundOperations(jsonObject);
                            } catch (Exception e) {
                                redisQueueService.sendMessage(Constant.RECEIVEREPEAT, mge);
                                // 取消或停止任务：当一个线程正在执行某个长时间运行的任务时，可以通过调用 interrupt() 方法来请求该线程停止执行。
                                 Thread.currentThread().interrupt();
                            }
                        });
                    }
                }
            }
        }
    }


    @Async
    public void receiveRepeat() {
        System.out.println("开始消费消息多线程失败的信息");
        // 开启线程池，如果有数据就启动线程消费
        final int[] tolot = {0};
        int corePoolSize = 10; // 核心线程数
        int maximumPoolSize = 50; // 最大线程数
        long keepAliveTime = 60L; // 空闲线程存活时间（秒）
        TimeUnit unit = TimeUnit.SECONDS; // 时间单位
        BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>(200); // 工作队列
        ThreadFactory threadFactory = Executors.defaultThreadFactory(); // 线程工厂
        RejectedExecutionHandler handler = new ThreadPoolExecutor.AbortPolicy(); // 拒绝策略
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                corePoolSize,
                maximumPoolSize,
                keepAliveTime,
                unit,
                workQueue,
                threadFactory,
                handler
        );
        while (true) {
            synchronized(this){
                //取出消息
                boolean message = redisQueueService.isQueueEmpty(Constant.RECEIVEREPEAT);
                if (message) {
//                System.out.println("消息队列为空");
                    try {
                        Thread.sleep(1000 * 5);
                    } catch (InterruptedException e) {
                    }
                } else {

                    // 检查工作队列是否已满
                    if (workQueue.size() >= workQueue.remainingCapacity()) {
                        System.out.println("工作队列已满，无法提交新任务");
                        // 处理队列已满的情况，例如记录日志或采取其他措施
                    } else {
                        executor.submit(() -> {
                            Object mge = redisQueueService.receiveMessage(Constant.RECEIVEREPEAT);
                            if (mge == null) {
                                return;
                            }
                            tolot[0]++;
                            int finalTolot = tolot[0];
                            try {
                                System.out.println("ZAICI消费消息------------：第（" + finalTolot + "）条");
                                JSONObject jsonObject = (JSONObject) JSON.toJSON(mge);
                                monitorBusiness.inboundOperations(jsonObject);
                            } catch (Exception e) {
//                                redisQueueService.sendMessage(Constant.RECEIVEREPEAT, mge);
                                // 取消或停止任务：当一个线程正在执行某个长时间运行的任务时，可以通过调用 interrupt() 方法来请求该线程停止执行。
                                Thread.currentThread().interrupt();
                            }
                        });
                    }
                }
            }
        }
    }

}
